package com.atguigu.tingshu.search.mq;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.search.service.SearchService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

@Component
@Slf4j
public class SearchReceiver {

    @Autowired
    private SearchService searchService ;

    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_UPPER)
    public void albumUpper(ConsumerRecord<String , String> consumerRecord) {
        String albumId = consumerRecord.value();
        if(!StringUtils.isEmpty(albumId)) {
            searchService.upperAlbum(Long.parseLong(albumId));
            log.info("专辑上架成功了,albumId: {} " , albumId);
        }
    }

    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_LOWER)
    public void albumLower(ConsumerRecord<String , String> consumerRecord) {
        String albumId = consumerRecord.value();
        if(!StringUtils.isEmpty(albumId)) {
            searchService.lowerAlbum(Long.parseLong(albumId));
            log.info("专辑下架成功了,albumId: {} " , albumId);
        }
    }

    @KafkaListener(topics = KafkaConstant.QUEUE_SYNC_ES_ALBUM_STAT)
    public void albumStatSync(ConsumerRecord<String , String> consumerRecord) {
        String value = consumerRecord.value();
        if(!StringUtils.isEmpty(value)) {
            AlbumStat albumStat = JSON.parseObject(value, AlbumStat.class);
            searchService.albumStatSync(albumStat) ;
            log.info("专辑的统计数据更新完毕, albumId: {}" , albumStat.getAlbumId());
        }
    }

}
